package com.wtz.kafka;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.twitter.bijection.Injection;
import com.twitter.bijection.avro.GenericAvroCodecs;
import com.wtz.kafka.entity.RandomValue;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @author tiezhu
 * @date 2021/3/29 星期一
 * Company dtstack
 */
public class KafkaAvro {

    private static final String BROKER_LIST = "kudu1:9092";

    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) throws InterruptedException {
        KafkaProducer<String, byte[]> kafkaProducer = buildKafkaProducer();
        String schema = readJsonAvroSchema("/Users/wtz/dtstack/json/stream/avroDemoEnums.json");

        JSONObject jsonSchema = (JSONObject) JSONObject.parse(schema);
        JSONArray fields = (JSONArray) jsonSchema.get("fields");

        Schema.Parser parser = new Schema.Parser();
        Schema arvoSchema = parser.parse(schema);
        Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(arvoSchema);
        for (int i = 0; i < 1000; i++) {
//            new GenericData.EnumSymbol(schema);
            GenericRecord genericRecord = new GenericData.Record(arvoSchema);
            JSONObject parse = (JSONObject) JSONObject.parse(RandomValue.buildJsonData(i));

            fields.forEach(
               item -> {
                   JSONObject jsonObject = (JSONObject) item;
               }
            );

            genericRecord.put("id", String.valueOf(i));
            genericRecord.put("name", "aa" + i);
            genericRecord.put("message", "bb" + i);

            byte[] bytes = recordInjection.apply(genericRecord);
            ProducerRecord<String, byte[]> tiezhuInTwo = new ProducerRecord<>("tiezhu_in_two", bytes);
            kafkaProducer.send(tiezhuInTwo);
            TimeUnit.SECONDS.sleep(1);
        }
    }

    public static KafkaProducer<String, byte[]> buildKafkaProducer() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());

        return new KafkaProducer<>(properties);
    }

    public static String readJsonAvroSchema(String filePath) {
        File file = new File(filePath);
        StringBuilder builder = new StringBuilder();
        if (!file.exists()) {
            throw new RuntimeException("file not exist! File path: " + filePath);
        }
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8))) {
            boolean isRunning = true;
            while (isRunning) {
                String line = reader.readLine();
                if (line == null) {
                    isRunning = false;
                } else {
                    builder.append(line);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return builder.toString();
    }
}
